package com.rem.flink.flink8State;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.util.Collector;

/**
 * 按键分区状态 ListState
 * 将两条流的笛卡尔积输出
 *
 * @author Rem
 * @date 2022-10-24
 */

public class ListStateTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<Tuple3<String, String, Long>> stream1 = env.fromElements(
                Tuple3.of("order-1", "app", 1000L),
                Tuple3.of("order-2", "app", 2000L)
        ).assignTimestampsAndWatermarks(WatermarkStrategy.<Tuple3<String, String, Long>>forMonotonousTimestamps()
                .withTimestampAssigner((element, recordTimestamp) -> element.f2));


        SingleOutputStreamOperator<Tuple3<String, String, Long>> stream2 = env.fromElements(
                Tuple3.of("order-1", "third", 4000L),
                Tuple3.of("order-2", "third", 5000L),
                Tuple3.of("order-1", "third", 3000L)
        ).assignTimestampsAndWatermarks(WatermarkStrategy.<Tuple3<String, String, Long>>forMonotonousTimestamps()
                .withTimestampAssigner((element, recordTimestamp) -> element.f2));


        stream1.keyBy(r -> r.f0)
                .connect(stream2.keyBy(r2 -> r2.f0))
                .process(new CoProcessFunction<Tuple3<String, String, Long>, Tuple3<String, String, Long>, String>() {

                    ListState<Tuple2<String, Long>> stream1;
                    ListState<Tuple2<String, Long>> stream2;

                    @Override
                    public void open(Configuration parameters) {
                        stream1 = super.getRuntimeContext().getListState(new ListStateDescriptor<Tuple2<String, Long>>("stream1", Types.TUPLE(Types.STRING, Types.LONG)));
                        stream2 = super.getRuntimeContext().getListState(new ListStateDescriptor<Tuple2<String, Long>>("stream1", Types.TUPLE(Types.STRING, Types.LONG)));

                    }

                    /**
                     * 在第一条流中加入第二条流的数据
                     * @param stream1Tuple
                     * @param ctx
                     * @param out
                     * @throws Exception
                     */
                    @Override
                    public void processElement1(Tuple3<String, String, Long> stream1Tuple, Context ctx, Collector<String> out) throws Exception {
                        //获取另外一条输出流 配对输出
                        for (Tuple2<String, Long> stream2Tuple : stream2.get()) {
                            out.collect(stream1Tuple.f0 + " " + stream1Tuple.f2 + " ===> " + stream2Tuple);
                        }
                        stream1.add(Tuple2.of(stream1Tuple.f0, stream1Tuple.f2));
                    }

                    /**
                     * 在第二条流中加入第一条流中的数据
                     * @param stream2Tuple
                     * @param ctx
                     * @param out
                     * @throws Exception
                     */
                    @Override
                    public void processElement2(Tuple3<String, String, Long> stream2Tuple, Context ctx, Collector<String> out) throws Exception {
                        //获取另外一条输出流 配对输出
                        for (Tuple2<String, Long> stream1Tuple : stream1.get()) {
                            out.collect(stream1Tuple + "===> " + stream2Tuple.f0 + " " + stream2Tuple.f2);
                        }
                        stream2.add(Tuple2.of(stream2Tuple.f0, stream2Tuple.f2));
                    }
                }).print();

        env.execute();


    }
}
